package demo.mq.rabbit;

import com.rabbitmq.client.*;
import com.rabbitmq.client.AMQP.BasicProperties;
import demo.java.lang.ThreadDemo;
import org.junit.Test;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.io.IOException;
import java.nio.charset.StandardCharsets;
import java.util.Random;
import java.util.concurrent.TimeoutException;

/**
 * AMQP协议 - 核心概念
 * <ol>
 * <li>server：又称broker，接受客户端连接，实现AMQP实体服务。
 * <li>connection：连接和具体broker网络连接。
 * <li>channel：网络信道，几乎所有操作都在channel中进行，channel是消息读写的通道。客户端可以建立多个channel，
 * 每个channel表示一个会话任务。
 * <li>message：消息，服务器和应用程序之间传递的数据，由properties和body组成。properties可以对消息进行修饰，
 * 比如消息的优先级，延迟等高级特性；body是消息实体内容。
 * <li>Virtual host：虚拟主机，用于逻辑隔离，最上层消息的路由。一个Virtual host可以若干个Exchange和Queue，
 * 同一个Virtual host不能有同名的Exchange或Queue。
 * <li>Exchange：交换机，接受消息，根据路由键转发消息到绑定的队列上。
 * <li>banding：Exchange和Queue之间的虚拟连接，binding中可以包括routing key
 * <li>routing key：一个路由规则，虚拟机根据他来确定如何路由 一条消息。
 * <li>Queue：消息队列，用来存放消息的队列。
 * </ol>
 * <h2>交换机的类型</h2>，direct、topic、fanout、headers，durability（是否需要持久化true需要）auto
 * delete当最后一个绑定Exchange上的队列被删除Exchange也删除。
 * <li>Direct Exchange,所有发送到Direct Exchange的消息被转发到RouteKey 中指定的Queue, Direct
 * Exchange可以使用默认的默认的Exchange （default Exchange），默认的Exchange会绑定所有的队列，
 * 所以Direct可以直接使用Queue名（作为routing key ）绑定。或者消费者和生产者的routing key完全匹配。
 * <li>Topic Exchange,是指发送到Topic Exchange的消息被转发到所有关心的Routing key中指定topic的Queue上。
 * Exchange 将routing key和某Topic进行模糊匹配，此时队列需要绑定一个topic。
 * 所谓模糊匹配就是可以使用通配符，“#”可以匹配一个或多个词，“”只匹配一个词比如“log.#”可以匹配“log.info.test” "log.
 * "就只能匹配log.error。
 * <li>Fan out Exchange:不处理路由键，只需简单的将队列绑定到交换机上。发送到改交换机上的消息都会被发送到与该交换机绑定的队列上。Fan
 * out转发是最快的。
 */
public class RabbitMqDemo {

    private static Logger logger = LoggerFactory.getLogger(RabbitMqDemo.class);

    /**
     * <h1>死信队列</h1>DLX，Dead-Letter-Exchange 利用DLX，当消息在一个队列中变成死信（dead
     * message，就是没有任何消费者消费）之后，他能被重新publish到另一个Exchange，这个Exchange就是DLX。 消息变为死信的几种情况：
     * <li>消息被拒绝（basic.reject/basic.nack）同时requeue=false（不重回队列）
     * <li>TTL过期
     * <li>队列达到最大长度
     * <p>
     * DLX也是一个正常的Exchange，和一般的Exchange没有任何的区别，他能在任何的队列上被指定，实际上就是设置某个队列的属性。
     * 当这个队列出现死信的时候，RabbitMQ就会自动将这条消息重新发布到Exchange上去，进而被路由到另一个队列。
     * 可以监听这个队列中的消息作相应的处理，这个特性可以弥补rabbitMQ以前支持的immediate参数的功能。
     * 
     * <h1>死信队列的设置</h1>
     * 
     * 设置Exchange和Queue，然后进行绑定 Exchange: dlx.exchange(自定义的名字)
     * 
     * queue: dlx.queue（自定义的名字）
     * 
     * routingkey: #（#表示任何routingkey出现死信都会被路由过来）
     * 
     * 然后正常的声明交换机、队列、绑定，只是我们在队列上加上一个参数：
     * arguments.put("x-dead-letter-exchange","dlx.exchange");
     */
    public void deadLetterExchange() {
        try {
            // 获取到连接以及mq通道
            Connection connection = getConnection();
            // 从连接中创建通道
            Channel channel = connection.createChannel();
            // 声明（创建）队列
            channel.queueDeclare(QUEUE_NAME, false, false, false, null);

            channel.exchangeDeclare("dlx", BuiltinExchangeType.HEADERS);
        } catch (IOException | TimeoutException e) {
            e.printStackTrace();
        }
    }

    /**
     * 消息的确认，指生产者收到投递消息后，如果Broker收到消息就会给我们 的生产者一个应答，生产者接受应答来确认broker是否收到消息。
     * <p>
     * 如何实现confirm确认消息。
     * <li>：channel.confirmSelect()
     * <li>在channel上添加监听：addConfirmListener，监听成功和失败的结果，具体结果对消息进行重新发送或者记录日志。
     * 
     * <h1>TTL队列/消息</h1> “TTL time to live 生存时间。 支持消息的过期时间，在消息发送时可以指定。
     * 支持队列过期时间，在消息入队列开始计算时间，只要超过了队列的超时时间配置，那么消息就会自动的清除。
     * 
     * 
     */
    public void channelProducerDemo() {
        /*
         * The empty string denotes the default or nameless exchange: messages are
         * routed to the queue with the name specified by routingKey
         */
        String exchange = "";
        // 队列名
        String queueName = QUEUE_NAME;
        String message = "大江东去，浪淘尽。";
        try {
            // 获取到连接以及mq通道
            Connection connection = getConnection();
            // 从连接中创建通道
            Channel channel = connection.createChannel();
            // 在Channel上开启确认模式
            channel.confirmSelect();
            // 在channel上添加监听：
            channel.addConfirmListener(new ConfirmListener() {
                @Override
                public void handleNack(long deliveryTag, boolean multiple) throws IOException {
                    logger.warn("deliveryTag: {} , multiple: {}", deliveryTag, multiple);
                }

                @Override
                public void handleAck(long deliveryTag, boolean multiple) throws IOException {
                    logger.info("deliveryTag: {} , multiple: {}", deliveryTag, multiple);
                }
            });
            // 发送消息的时候当Exchange不存在或者指定的路由key路由找不到，如果我们需要监听这种不可到达的消息，就要使用Return Listener！
            // Mandatory 设置为true则会监听器会接受到路由不可达的消息，然后处理。如果设置为false，broker将会自动删除该消息。
            channel.addReturnListener(new ReturnListener() {
                @Override
                public void handleReturn(int replyCode, String replyText, String exchange, String routingKey,
                        AMQP.BasicProperties properties, byte[] body) throws IOException {
                    logger.warn("send msg failed, replyCode: {} , replyText:{} , exchange:{} ,routingKey:{} ",
                            replyCode, replyText, exchange, routingKey);

                }
            });

            // make sure that the queue will survive a RabbitMQ node restart.
            boolean durable = true;
            // 声明队列
            channel.queueDeclare(queueName, durable, false, false, null);

            /*
             * mark our messages as persistent. Marking messages as persistent doesn't fully
             * guarantee that a message won't be lost. Although it tells RabbitMQ to save
             * the message to disk, there is still a short time window when RabbitMQ has
             * accepted a message and hasn't saved it yet. Also, RabbitMQ doesn't do
             * fsync(2) for every message -- it may be just saved to cache and not really
             * written to the disk.
             */
            BasicProperties props = MessageProperties.PERSISTENT_TEXT_PLAIN;

            String routingKey = "task_queue";
            //
            long sequenceNumber = channel.getNextPublishSeqNo();
            logger.info("publishSeqNo = {}", sequenceNumber);
            channel.basicPublish(exchange, routingKey, props, message.getBytes());
            // uses a 5 second timeout
            channel.waitForConfirmsOrDie(5_000);
        } catch (IOException e) {
            logger.error("", e);
        } catch (TimeoutException e) {
            logger.error("", e);
        } catch (InterruptedException e) {
            e.printStackTrace();
        }

    }

    /**
     * <h1>消费端限流</h1> rabbitMQ提供了一种qos（服务质量保证）的功能，即非自动确认消息的前提下，
     * 如果有一定数目的消息（通过consumer或者Channel设置qos）未被确认，不进行新的消费。
     * 
     * <h1>消费端ack与重回队列</h1> 消费端进行消费的时候，如果由于业务异常我们可以进行日志的记录，然后进行补偿！
     * 如果由于服务器宕机等严重问题，那我们就需要手动进行ack保证消费端的消费成功！
     * 重回队列就是为了对没有处理成功的消息，把消息重新投递给broker！实际应用中一般都不开启重回队列。
     * 
     * 
     */
    @Test
    public void channelConsumer() {
        try {
            // 获取到连接以及mq通道
            Connection connection = getConnection();
            // 从连接中创建通道
            Channel channel = connection.createChannel();

            // 单条消息的大小限制。0就是不限制，一般都是不限制。
            int prefetchSize = 0;
            // 设置一个固定的值，告诉rabbitMQ不要同时给一个消费者推送多于N个消息，即一旦有N个消息还没有ack，则consumer将block掉，直到有消息ack
            int prefetchCount = 1;
            // 是否将上面的设置用于channel，也是就是说上面设置的限制是用于channel级别的还是consumer的级别的。
            boolean global = true;
            // 消费端限流
            channel.basicQos(prefetchSize, prefetchCount, global);

            channel.queueDeclare(QUEUE_NAME, true, false, false, null);
        } catch (IOException e) {
            logger.error("", e);
        } catch (TimeoutException e) {
            logger.error("", e);
        }

    }

    /**
     * 获取MQ的连接
     * 
     * @return
     * @throws IOException
     * @throws TimeoutException
     */
    public static Connection getConnection() throws IOException, TimeoutException {
        // 定义连接工厂
        ConnectionFactory factory = new ConnectionFactory();
        // 设置服务地址
        factory.setHost("192.168.3.161");
        // 端口
        factory.setPort(5672);
        // 设置账号信息，用户名、密码、vhost
        factory.setVirtualHost(ConnectionFactory.DEFAULT_VHOST);
        factory.setUsername("guest");
        factory.setPassword("guest");
        // 通过工程获取连接
        Connection connection = factory.newConnection();
        return connection;
    }

    private final static String QUEUE_NAME = "q_test_01";

    /**
     * 生产者发送消息到队列
     * 
     * @throws TimeoutException
     * @throws IOException
     * 
     * @throws Exception
     */
    @Test
    public void testSendMsg() throws IOException, TimeoutException {
        sendMsg("百舸争流");
    }

    /**
     * 生产者发送消息到队列
     * 
     * @throws TimeoutException
     * @throws IOException
     * 
     * @throws Exception
     */
    public static void sendMsg(String message) {
        if (message == null) {
            return;
        }
        try ( // 获取到连接以及mq通道
                Connection connection = getConnection();
                // 从连接中创建通道
                Channel channel = connection.createChannel();) {

            // 在Channel上开启确认模式
            channel.confirmSelect();
            // 在channel上添加监听：
            channel.addConfirmListener(new ConfirmListener() {
                @Override
                public void handleNack(long deliveryTag, boolean multiple) throws IOException {
                    logger.warn("deliveryTag: {} , multiple: {}", deliveryTag, multiple);
                }

                @Override
                public void handleAck(long deliveryTag, boolean multiple) throws IOException {
                    logger.info("deliveryTag: {} , multiple: {}", deliveryTag, multiple);
                }
            });
            // 发送消息的时候当Exchange不存在或者指定的路由key路由找不到，如果我们需要监听这种不可到达的消息，就要使用Return Listener！
            // Mandatory 设置为true则会监听器会接受到路由不可达的消息，然后处理。如果设置为false，broker将会自动删除该消息。
            channel.addReturnListener(new ReturnListener() {
                @Override
                public void handleReturn(int replyCode, String replyText, String exchange, String routingKey,
                        AMQP.BasicProperties properties, byte[] body) throws IOException {
                    logger.warn("send msg failed, replyCode: {} , replyText:{} , exchange:{} ,routingKey:{} ",
                            replyCode, replyText, exchange, routingKey);

                }
            });
            // 声明（创建）队列
            channel.queueDeclare(QUEUE_NAME, false, false, false, null);
            logger.info("发送消息：{}", message);
            // 消息内容
            channel.basicPublish("", QUEUE_NAME, null, message.getBytes(StandardCharsets.UTF_8));
            // 消息内容
            channel.basicPublish("", QUEUE_NAME, null, message.getBytes(StandardCharsets.UTF_8));

        } catch (IOException | TimeoutException e) {
            logger.error("", e);
        }
    }

    public void comsumer() throws IOException, TimeoutException {
        // 获取到连接以及mq通道
        Connection connection = getConnection();
        // 从连接中创建通道
        Channel channel = connection.createChannel();

        // 声明队列
        channel.queueDeclare(QUEUE_NAME, false, false, false, null);
        System.out.println(" [*] Waiting for messages. To exit press CTRL+C");
        // 定义队列的消费者
        DefaultConsumer consumer = new DefaultConsumer(channel);
        // 监听队列
        channel.basicConsume(QUEUE_NAME, true, consumer);
        DeliverCallback deliverCallback = new DeliverCallback() {
            @Override
            public void handle(String consumerTag, Delivery delivery) throws IOException {
                String message = new String(delivery.getBody(), "UTF-8");
                System.out.println(" [x] Received '" + message + "'");
            }
        };
        channel.basicConsume(QUEUE_NAME, true, deliverCallback, consumerTag -> {
        });
    }

    /**
     * 一个生成者对应一个消费者。 生产者发送消息到queue里，消费者从queue里消费消息
     * <li>A program that sends messages is a producer :
     * <li>A queue is the name for a post box which lives inside RabbitMQ. Although
     * messages flow through RabbitMQ and your applications, they can only be stored
     * inside a queue. A queue is only bound by the host's memory & disk limits,
     * it's essentially a large message buffer. Many producers can send messages
     * that go to one queue, and many consumers can try to receive data from one
     * queue. This is how we represent a queue:
     * <li>A consumer is a program that mostly waits to receive messages:
     * 
     * @see https://www.rabbitmq.com/tutorials/tutorial-one-java.html
     */
    @Test
    public void simplestQueueDemo() {
        String queueName = "q_test_01";
        ConnectionFactory factory = new ConnectionFactory();
        factory.setHost("localhost");

        Thread producer = new Thread(new Runnable() {
            @Override
            public void run() {

                try (Connection connection = factory.newConnection(); Channel channel = connection.createChannel()) {
                    boolean durable = false;
                    boolean autoDelete = false;
                    channel.queueDeclare(queueName, durable, false, autoDelete, null);
                    for (int i = 0; i < 10; i++) {
                        String message = i + " 大江东去，浪淘尽。";
                        channel.basicPublish("", queueName, null, message.getBytes());
                        logger.info("Send Msg: {}", message);
                        ThreadDemo.safeSleep(500);
                    }
                } catch (IOException e) {
                    e.printStackTrace();
                } catch (TimeoutException e) {
                    e.printStackTrace();
                }
            }
        }, "producer");

        Thread consumer = new Thread(new Runnable() {
            @Override
            public void run() {
                try {
                    Connection connection = factory.newConnection();
                    Channel channel = connection.createChannel();
                    channel.queueDeclare(queueName, false, false, false, null);

                    DeliverCallback deliverCallback = new DeliverCallback() {
                        @Override
                        public void handle(String consumerTag, Delivery message) throws IOException {
                            String msg = new String(message.getBody(), "UTF-8");
                            logger.info("Received Msg: {} ", msg);
                        }
                    };

                    String consumerTag = channel.basicConsume(queueName, true, deliverCallback, new CancelCallback() {
                        @Override
                        public void handle(String consumerTag) throws IOException {
                            logger.info("consumerTag : {} ", consumerTag);
                        }
                    });
                    logger.info("consumer [{}] Waiting for messages", consumerTag);
                } catch (IOException e) {
                    e.printStackTrace();
                } catch (TimeoutException e) {
                    e.printStackTrace();
                }

            }
        }, "consumer");

        producer.start();
        ThreadDemo.safeSleep(3 * 1000);
        consumer.start();
        ThreadDemo.safeSleep(3 * 60 * 1000);
    }

    /**
     * 多个消费者共同处理消息。Round-robin dispatching ，One of the advantages of using a Task
     * Queue is the ability to easily parallelise work. If we are building up a
     * backlog of work, we can just add more workers and that way, scale easily.
     * 
     * @see https://www.rabbitmq.com/tutorials/tutorial-two-java.html
     */
    @Test
    public void workQueueDemo() {
        String queueName = "task_queue";
        ConnectionFactory factory = new ConnectionFactory();
        factory.setHost("localhost");
        /**
         * 消息持久化： 1、 设置队列 durable = true
         * 2、发送的消息指定属性MessageProperties.PERSISTENT_TEXT_PLAIN
         */
        Thread producer = new Thread(new Runnable() {
            @Override
            public void run() {

                try (Connection connection = factory.newConnection(); Channel channel = connection.createChannel()) {
                    boolean durable = true;
                    boolean autoDelete = false;
                    channel.queueDeclare(queueName, durable, false, autoDelete, null);
                    for (int i = 0; i < 10; i++) {
                        String message = i + " 滚滚长江东逝水!";
                        channel.basicPublish("", queueName, MessageProperties.PERSISTENT_TEXT_PLAIN,
                                message.getBytes());
                        logger.info("Send Msg: {}", message);
                        ThreadDemo.safeSleep(20);
                    }
                } catch (IOException e) {
                    e.printStackTrace();
                } catch (TimeoutException e) {
                    e.printStackTrace();
                }
            }
        }, "producer");

        Runnable consumerTask = new Runnable() {
            @Override
            public void run() {
                try {
                    Connection connection = factory.newConnection();
                    Channel channel = connection.createChannel();
                    boolean durable = true;
                    boolean autoDelete = false;
                    channel.queueDeclare(queueName, durable, false, autoDelete, null);
                    // accept only one unack-ed message at a time (see below)
                    channel.basicQos(1);

                    DeliverCallback deliverCallback = new DeliverCallback() {
                        @Override
                        public void handle(String consumerTag, Delivery message) throws IOException {
                            String msg = new String(message.getBody(), "UTF-8");
                            logger.info("消费者1 Received Msg: {} ", msg);
                            ThreadDemo.safeSleep(1000);
                            /**
                             * 消费消息成功后，发送acknowledgement。 If a consumer dies (its channel is closed,
                             * connection is closed, or TCP connection is lost) without sending an ack,
                             * RabbitMQ will understand that a message wasn't processed fully and will
                             * re-queue it. If there are other consumers online at the same time, it will
                             * then quickly redeliver it to another consumer. That way you can be sure that
                             * no message is lost, even if the workers occasionally die.
                             */
                            channel.basicAck(message.getEnvelope().getDeliveryTag(), false);
                            logger.info("消费者1 [x] Done");
                        }
                    };
                    boolean autoAck = false;
                    String consumerTag = channel.basicConsume(queueName, autoAck, deliverCallback, e -> {
                    });
                    logger.info("consumer [{}] Waiting for messages", consumerTag);
                } catch (IOException e) {
                    e.printStackTrace();
                } catch (TimeoutException e) {
                    e.printStackTrace();
                }

            }
        };

        Thread consumer1 = new Thread(consumerTask, "consumer1");
        Thread consumer2 = new Thread(consumerTask, "consumer2");

        producer.start();
        ThreadDemo.safeSleep(6 * 1000);
        consumer1.start();
        consumer2.start();
        ThreadDemo.safeSleep(3 * 60 * 1000);
    }


    @Test
    public void directExchangeDemo() {
        String exchangeName = "direct_logs";
        String routingKey = "0";
        ConnectionFactory factory = new ConnectionFactory();
        factory.setHost("localhost");
        /**
         * 消息持久化： 1、 设置队列 durable = true
         * 2、发送的消息指定属性MessageProperties.PERSISTENT_TEXT_PLAIN
         */
        Thread producer = new Thread(new Runnable() {
            @Override
            public void run() {
                try (Connection connection = factory.newConnection();
                     Channel channel = connection.createChannel()) {
                    channel.exchangeDeclare(exchangeName, "direct");
                    for (int i = 0; i < 100; i++) {
                        String message = i + " 滚滚长江东逝水!";
                        String routingKeyVar = String.valueOf(i % 2);
                        channel.basicPublish(exchangeName, routingKeyVar, null, message.getBytes("UTF-8"));
                        logger.info(" [x] Sent '" + message + "'");
                        ThreadDemo.safeSleep(300);
                    }
                } catch (IOException e) {
                    e.printStackTrace();
                } catch (TimeoutException e) {
                    e.printStackTrace();
                }
            }
        }, "producer");
        Random random = new Random();
        Runnable consumerTask = new Runnable() {
            @Override
            public void run() {
                int n = random.nextInt(100);
                try {
                    Connection connection = factory.newConnection();
                    Channel channel = connection.createChannel();
                    channel.exchangeDeclare(exchangeName, "direct");
                    /*
                     * queueDeclare() create a non-durable, exclusive, autoDelete queue with a
                     * generated name:
                     */
                    String queueName = channel.queueDeclare().getQueue();
                    // 绑定 queue 和 exchange
                    channel.queueBind(queueName, exchangeName, "");

                    // accept only one unack-ed message at a time (see below)
                    channel.basicQos(1);
                    logger.info("n +  [*] Waiting for messages. To exit press CTRL+C");

                    DeliverCallback deliverCallback = (consumerTag, delivery) -> {
                        String message = new String(delivery.getBody(), "UTF-8");
                        System.out.println(
                                " [x] Received '" + delivery.getEnvelope().getRoutingKey() + "':'" + message + "'");
                        boolean ok = random.nextBoolean();
                        if (ok) {
                            channel.basicAck(delivery.getEnvelope().getDeliveryTag(), false);
                            System.out.println("消费成功");
                        } else {
                            System.out.println("消费失败");
                            channel.basicNack(delivery.getEnvelope().getDeliveryTag(), false, true);
                        }
                    };
                    // acknowledgment is covered below
                    boolean autoAck = false;
                    String consumerTag = channel.basicConsume(queueName, autoAck, deliverCallback,
                            new CancelCallback() {
                                @Override
                                public void handle(String consumerTag) throws IOException {
                                    logger.info("CancelCallback , consumerTag : {} ", consumerTag);
                                    ThreadDemo.safeSleep(500);
                                }
                            });
                    logger.info("consumer [{}] Waiting for messages", consumerTag);
                } catch (IOException e) {
                    e.printStackTrace();
                } catch (TimeoutException e) {
                    e.printStackTrace();
                }

            }
        };

        producer.start();
        ThreadDemo.safeSleep(3 * 1000);
        Thread consumer1 = new Thread(consumerTask, "consumer1");
        consumer1.start();
        ThreadDemo.safeSleep(3 * 1000);
        Thread consumer2 = new Thread(consumerTask, "consumer2");
        consumer2.start();
        ThreadDemo.safeSleep(3 * 60 * 1000);
    }

}
